package org.rency.crawler.scheduler.listeners;

import javax.annotation.Resource;
import org.rency.common.utils.converts.JsonConvert;
import org.rency.crawler.common.CrawlerConstants;
import org.rency.crawler.domain.FetchExtension;
import org.rency.crawler.domain.FetchRequest;
import org.rency.crawler.domain.queue.FetchQueue;
import org.rency.crawler.fetch.helper.BuilderHelper;
import org.rency.crawler.fetch.processor.FetchProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.util.Pair;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

/**
 * 描述: ${TODO}<br>
 * 最近修改时间:2017/11/29 17:43<br>
 *
 * @author CaiyuRen
 * @version V1.0
 * @since 2017/11/29 17:43
 */
@Component
public class FetchQueueListener implements CrawlerConstants{

  private static final Logger logger = LoggerFactory.getLogger(FetchQueueListener.class);

  @Resource
  private FetchProcessor     fetchProcessor;

  @Resource(name = "listenerTaskScheduler")
  private ThreadPoolTaskExecutor taskExecutor;

  @KafkaListener(id = "fetchQueue", topics = QUEUE_FETCH_NAME)
  public void listener(final String fetchQueue){
    logger.debug("监听待抓取队列:{}",fetchQueue);
    taskExecutor.submit(new Runnable() {
      @Override
      public void run() {
        FetchQueue queue = JsonConvert.toObject(fetchQueue,FetchQueue.class);
        Pair<FetchRequest, FetchExtension> pair = BuilderHelper.build(queue);
        fetchProcessor.fetch(pair.getFirst(),pair.getSecond());
      }
    });
  }
}
